package com.fxkj.cloud.drms.infra.mq.mqtt.sender;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fxkj.cloud.dbs.common.model.ControlOrder;
import com.fxkj.cloud.dbs.common.model.DeviceBasicInfo;
import com.fxkj.cloud.dbs.common.model.alarmItem.ProductAlarmItemCreate;
import com.fxkj.cloud.dbs.common.model.linkage.KvDTO;
import com.fxkj.cloud.drms.domain.mqtt.CommandAndQuerySender;
import com.fxkj.cloud.drms.domain.mqtt.CommandDomain;
import com.fxkj.cloud.drms.domain.mqtt.DeviceControlOrderCommand;
import com.fxkj.cloud.drms.domain.mqtt.QueryDomain;
import com.fxkj.cloud.drms.infra.mq.mqtt.MqttConstants;
import com.fxkj.cloud.drms.mqtt.MqttClient;
import com.fxkj.common.utils.JacksonUtil;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.MoreExecutors;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Set;

/**
 * <p>description: </p>
 * @author ww
 * @date 2025/3/25 10:20
 */
@Component
public class MqttSender implements CommandAndQuerySender {
    private final MqttClient mqttClient;


    public MqttSender(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }

    @Override
    public void commandSend(CommandDomain commandDomain) {
        MqttCommand mqttCommand = MqttCommand.create(commandDomain);
        String serialNo = commandDomain.getDeviceBasicInfo().getSerialNo();
        //mqtt推送消息
        mqttClient.publish(MqttConstants.Topic.Send.COMMAND_SEND.replace("+", serialNo), convertMessageToByteBuf(mqttCommand.toJsonString()));
    }

    @Override
    public void commandSendBatch(List<CommandDomain> commandDomains) {
        Futures.submit(()->{
            commandDomains.forEach(commandDomain -> {
                MqttCommand mqttCommand = MqttCommand.create(commandDomain);
                String serialNo = commandDomain.getDeviceBasicInfo().getSerialNo();
                //mqtt推送消息
                mqttClient.publish(MqttConstants.Topic.Send.COMMAND_SEND.replace("+", serialNo), convertMessageToByteBuf(mqttCommand.toJsonString()));
            });
        }, MoreExecutors.directExecutor());
    }

    @Override
    public void querySend(QueryDomain queryDomain) {
        MqttQuery mqttQuery = MqttQuery.create(queryDomain);
        String serialNo = queryDomain.getDeviceBasicInfo().getSerialNo();
        //mqtt推送消息
        mqttClient.publish(MqttConstants.Topic.Send.QUERY_SEND.replace("+", serialNo), convertMessageToByteBuf(mqttQuery.toJsonString()));
    }

    @Override
    public void controlPlanSend(List<DeviceControlOrderCommand> deviceControlOrderCommands) {
        deviceControlOrderCommands.forEach(command -> {
            String serialNo = command.getDeviceSerialNo();
            //mqtt推送消息
            mqttClient.publish(MqttConstants.Topic.Send.COMMAND_SEND.replace("+", serialNo), convertMessageToByteBuf(MqttCommand.createControlPlan(command).toJsonString()));
        });
    }

    @Override
    public void alarmAttributeCreate(Set<String> deviceSerialNos, List<ProductAlarmItemCreate> productAlarmItemCreates) {
        try {
            for (String deviceSerialNo : deviceSerialNos) {
                mqttClient.publish(MqttConstants.Topic.Send.ALARM_ITEM_SEND.replace("+", deviceSerialNo), convertMessageToByteBuf(JacksonUtil.writeValueAsString(productAlarmItemCreates)));
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void commandSend(DeviceBasicInfo deviceBasicInfo, List<KvDTO> kvs) {
        MqttCommand mqttCommand = MqttCommand.create(kvs);
        String serialNo = deviceBasicInfo.getSerialNo();
        //mqtt推送消息
        mqttClient.publish(MqttConstants.Topic.Send.COMMAND_SEND.replace("+", serialNo), convertMessageToByteBuf(mqttCommand.toJsonString()));
    }

    private ByteBuf convertMessageToByteBuf(String content) {
        byte[] publishMsgBytes = content.getBytes();
        ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer(publishMsgBytes.length);
        byteBuf.writeBytes(publishMsgBytes);
        return byteBuf;
    }
}
